package com.example.demo;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * @author panghu
 * @version 1.0.0
 * @createTime 2023年09月09日 17:04:00
 * @Description
 */
public class CanalDemo {

    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
        // 创建canal客户端，单链接模式
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "example", "", "");
        while (true) {
            // 创建连接
            canalConnector.connect();
            // 订阅数据库
            canalConnector.subscribe("canal_demo.*");
            // 获取变更的所有数据信息 Message
            Message message = canalConnector.get(100);
            // 获取Entry集合，一条sql语句的执行，改动数据，将这些数据封装到Entry中
            List<CanalEntry.Entry> entries = message.getEntries();
            // 判断集合是否为空,如果为空,则等待一会继续拉取数据
            if (entries.size() <= 0) {
                System.out.println("当次抓取没有数据，休息一会.......");
                Thread.sleep(1000); //thead ---- binlog
            } else {
                // 遍历entries，单条解析
                for (CanalEntry.Entry entry : entries) {
                    //1.获取表名
                    String tableName = entry.getHeader().getTableName();
                    //2.获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();
                    //3.获取序列化后的数据
                    ByteString storeValue = entry.getStoreValue();
                    //4.判断当前entryType类型是否为ROWDATA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                        //5.反序列化数据
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        //6.获取当前事件的操作类型
                        CanalEntry.EventType eventType = rowChange.getEventType();
                        //7.获取数据集
                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
                        //8.遍历rowDataList，并打印数据集
                        for (CanalEntry.RowData rowData : rowDataList) {
                            // 获取修改前的数据
                            Map<String,Object> baforeMap  = new HashMap<>();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                baforeMap.put(column.getName(), column.getValue());
                            }

                            // 获取修改后的数据
                            Map<String,Object> afterMap = new HashMap<>();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterMap.put(column.getName(), column.getValue());
                            }

                            //数据打印
                            System.out.println("表名: " + tableName + ",操作类型: " + eventType);
                            System.out.println("改前: " + baforeMap);
                            System.out.println("改后: " + afterMap);
                        }
                    }
                }
            }
        }
    }
}



